package cn.net.yugu.doraemon.datahub.worker;

import cn.net.yugu.doraemon.datahub.annotation.DataHubHandler;
import cn.net.yugu.doraemon.datahub.annotation.DataHubListener;
import cn.net.yugu.doraemon.datahub.config.DataHubProperties;
import cn.net.yugu.doraemon.datahub.config.DataHubConfig;
import cn.net.yugu.doraemon.datahub.model.MsgConsumer;
import com.aliyun.datahub.client.DatahubClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

/**
 * @description:
 * @author: wanghongli
 * @create: 2022-07-17 11:02:41
 */
@Slf4j
public class RecordConsumeWorker {
    private static Map<String, MsgConsumer> dataWorkMap;
    private static Map<String, RecordConsumeWorkerThread>  dataListenerWorkerMap;

    public static synchronized void initWorkerResource(
            DatahubClient datahubClient,
            ApplicationContext context,
            DataHubProperties accountProperties) {
        Map<String, Object> beansMap = context.getBeansWithAnnotation(DataHubListener.class);
        if (CollectionUtils.isEmpty(beansMap)) {
            log.info("datahub:no listener need process");
            return;
        }

        if (dataListenerWorkerMap == null) {
            dataListenerWorkerMap = new HashMap<>();
        }
        if (dataWorkMap == null) {{
            dataWorkMap = new HashMap<>();
        }}
        for (Map.Entry<String, Object> entry : beansMap.entrySet()) {
            DataHubListener dataHubListener = entry.getValue().getClass().getAnnotation(DataHubListener.class);
            if (dataHubListener == null || StringUtils.isBlank(dataHubListener.projectName())) {
                log.error("dataHub:listener {} config error, projectName can not be blank", entry.getKey().getClass().getName());
                continue;
            }

            Method[] methods = entry.getValue().getClass().getDeclaredMethods();
            for (Method method : methods) {
                if (!method.isAnnotationPresent(DataHubHandler.class)) {
                    continue;
                }
                DataHubHandler dataHubHandler = method.getAnnotation(DataHubHandler.class);
                if (dataHubHandler == null || StringUtils.isBlank(dataHubHandler.topicName())) {
                    log.error("dataHub:listener {}'s handler {} config error topicName can not be blank", entry.getKey().getClass().getName(), method.getName());
                    continue;
                }

                DataHubConfig dataHubConfig = DataHubConfig.builder()
                        .projectName(dataHubListener.projectName())
                        .topicName(dataHubHandler.topicName())
                        .subId(dataHubHandler.subId())
                        .build();
                String workerKey = genWorkerKey(dataHubConfig.getProjectName(), dataHubConfig.getTopicName(), dataHubConfig.getSubId());
                Class<?> msgClazz = method.getParameterTypes()[0];
                RecordConsumeWorkerThread workerThread = new RecordConsumeWorkerThread(datahubClient,
                        accountProperties, workerKey, dataHubConfig, msgClazz);
                workerThread.setUncaughtExceptionHandler((t, e) -> log.error("datahub:worker thread {} uncaught exception", t.getName(), e));
                dataListenerWorkerMap.put(workerKey, workerThread);

                dataWorkMap.put(workerKey, MsgConsumer.builder().consumer(entry.getValue()).method(method).build());
            }
        }
    }

    public static synchronized void start() {
        if (CollectionUtils.isEmpty(dataListenerWorkerMap)) {
            return;
        }
        dataListenerWorkerMap.forEach((k, v) -> {
            v.start();
        });
    }

    public static synchronized void shutdown() {
        if (!CollectionUtils.isEmpty(dataListenerWorkerMap)) {
            dataListenerWorkerMap.forEach((k, v) -> v.shutdown());
        }
    }

    public static void invokeMethod(DataHubConfig config, Object record) {
        // 单线程消费，防止消费报错导致导致 offset 点异常.
        String workerKey = genWorkerKey(config.getProjectName(), config.getTopicName(), config.getSubId());
        MsgConsumer dataWork = dataWorkMap.get(workerKey);
        try {
            dataWork.getMethod().invoke(dataWork.getConsumer(), record);
        } catch (IllegalAccessException | InvocationTargetException e) {
            log.error("dataHub:consumer exception happened, project:{} topic:{} subId:{} data:{} exception:",
                    config.getProjectName(), config.getTopicName(), config.getSubId(), record, e);
        }
    }

    private static String genWorkerKey(String projectName, String topicName, String subId) {
        return (projectName + "-" + topicName + "-" + subId).intern();
    }
}
